package com.lifq.flink.streaming

import org.apache.flink.api.common.functions.AggregateFunction

/**
 * @Author: LiFuqiang
 * @Date: 2022/9/16 15:47
 * @Version 1.0 
 * @description
 */
class CustomAggFunction(threshold: Int) extends AggregateFunction[(String, Int), CustomAccumulator, (String, Int)] {
  override def createAccumulator(): CustomAccumulator = CustomAccumulator("", 0)

  override def add(value: (String, Int), accumulator: CustomAccumulator): CustomAccumulator = {
    if (value._2 >= threshold) {
      val count = accumulator.count + 1
      return CustomAccumulator(value._1, count)
    }
    CustomAccumulator(value._1, accumulator.count)
  }

  override def getResult(accumulator: CustomAccumulator): (String, Int) = {
    (accumulator.oil, accumulator.count)
  }

  override def merge(a: CustomAccumulator, b: CustomAccumulator): CustomAccumulator = {
    val count = a.count + b.count
    CustomAccumulator(a.oil, count)
  }
}

case class CustomAccumulator(oil: String, count: Int)
